package com.acme.exchange.service;

import java.io.IOException;
import java.math.BigDecimal;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.BlockingQueue;

import com.acme.exchange.domain.OpenBook;
import com.acme.exchange.domain.SymboledOrder;
import com.acme.exchange.domain.UserDataBasa;
import com.acme.exchange.domain.order.Order;
import com.acme.exchange.exception.IllegalExchangeMessageData;
import com.acme.exchange.network.MessageTransceiver;
import com.acme.exchange.network.proto.DefinedMessage;
import com.acme.exchange.network.proto.MessageConstructor;
import com.acme.exchange.network.proto.Protocol.ExchangeMessage;
import com.acme.exchange.network.proto.Protocol.ExchangeMessage.MessageType;
import com.acme.exchange.network.proto.Protocol.ExchangeMessage.OrderType;
import com.acme.exchange.network.proto.Protocol.ExchangeMessage.Side;

public class ClientMessageProcessor implements Runnable {

	private final BlockingQueue<DefinedMessage> clientMessages;
	private final LinkedList<UserDataBasa> tableOfUsers; // have to do in
															// blocking mode
	private Set<String> usersLogins;
	private final Set<SelectionKey> loginedClients;
	private MessageTransceiver mt;
	private final HashMap<String, OpenBook> exchange;

	public ClientMessageProcessor(BlockingQueue<DefinedMessage> q,
			LinkedList<UserDataBasa> tableOfUsers, Set<String> users,
			Set<SelectionKey> keys, HashMap<String, OpenBook> exchange) {
		clientMessages = q;
		this.tableOfUsers = tableOfUsers;
		usersLogins = users;
		loginedClients = keys;
		mt = new MessageTransceiver();
		this.exchange = exchange;
	}

	public void run() {
		DefinedMessage message;
		SymboledOrder orderTicker;
		ExchangeMessage em;
		try {
			while (true) {
				// trying to get message from the queue
				message = clientMessages.take();
				// check if the user is loginned correctly
				// if (!userIsLoginedCorrectly(message)) { // we don't work with
				// this user anymore

				synchronized (loginedClients) {

					if (!loginedClients.contains(message.getKey())) {
						// the first message need to be Logon

						userIsLoginedCorrectly(message); // check whether user
															// is logined
															// correctly
					} else {
						// checking if the order is correct
						try {
							orderTicker = messageProcess(message);

							// order is correct, so we generate a new order and
							// a report
							em = MessageConstructor
									.constructExecutionReportForNewOrder(
											orderTicker.getOrder(),
											orderTicker.getTicker());

							// send the report
							synchronized ((SocketChannel) message.getKey()
									.channel()) {
								mt.transmit(em.toByteArray(),
										(SocketChannel) message.getKey()
												.channel());
							}
							// trying to match the order
							processMatching(orderTicker.getOrder(),
									orderTicker.getTicker(), message.getKey());

						} catch (IllegalExchangeMessageData e) { // order is
																	// incorrect
							System.out.println(e.getMessage());
							// send a reject
							try {
								em = MessageConstructor
										.constructRejectMessage(e.getMessage());

								synchronized ((SocketChannel) message.getKey()
										.channel()) {
									mt.transmit(em.toByteArray(),
											(SocketChannel) message.getKey()
													.channel());
								}
							} catch (IOException e1) {
								System.out.println("transmition failed");
							}
						} catch (IOException e) {
							System.out.println("transmition failed");
						}
					}
				}

			}

		} catch (InterruptedException e) {
			System.out
					.println("couldn't take message from the queue with clients' messages");
		}

	}

	private SymboledOrder messageProcess(DefinedMessage message)
			throws IllegalExchangeMessageData {
		Order order;
		String ticker;
		Side s;
		int quantity;
		OrderType ot;
		BigDecimal price = new BigDecimal(0.0);
		int orderId;
		ExchangeMessage clientMessage;

		clientMessage = message.getMessage();

		if (!clientMessage.hasMessageType()) {
			throw new IllegalExchangeMessageData("message doesn't have a type");
		}
		if (clientMessage.getMessageType() != MessageType.NEW_ORDER) {
			throw new IllegalExchangeMessageData("message has a wrong type");
		}
		if (!clientMessage.hasClientOrderId()) {
			throw new IllegalExchangeMessageData(
					"the client order ID wasn't generated");
		}
		if (!clientMessage.hasSymbol()) {
			throw new IllegalExchangeMessageData("message miss a ticker");
		}
		if (!exchange.containsKey(clientMessage.getSymbol())) {
			throw new IllegalExchangeMessageData("message has a wrong ticker");
		}
		if (!clientMessage.hasSide()) {
			throw new IllegalExchangeMessageData(
					"message miss a type of operation (Sell/Buy)");
		}
		if (clientMessage.getSide() != Side.BUY
				&& clientMessage.getSide() != Side.SELL) {
			throw new IllegalExchangeMessageData(
					"message has a wrong type of operation");
		}
		if (!clientMessage.hasQuantity()) {
			throw new IllegalExchangeMessageData(
					"message miss a quantity of shares");
		}
		quantity = clientMessage.getQuantity();
		if (quantity <= 0 || quantity > 1000000) {
			throw new IllegalExchangeMessageData(
					"message has an illegal quantity of shares");
		}
		if (!clientMessage.hasOrderType()) {
			throw new IllegalExchangeMessageData(
					"message doesn't have an order types");
		}
		ot = clientMessage.getOrderType();
		if (ot != OrderType.LIMIT && ot != OrderType.MARKET) {
			throw new IllegalExchangeMessageData(
					"message has an illegal order type");
		}
		if (ot == OrderType.LIMIT) {
			if (!clientMessage.hasPrice()) {
				throw new IllegalExchangeMessageData(
						"message doesn't have a price");
			}
			try {
				price = new BigDecimal(clientMessage.getPrice());
				if (price.compareTo(new BigDecimal(0.0)) <= 0
						|| price.compareTo(new BigDecimal(10000000000.0)) == 1) {
					throw new NumberFormatException();
				}
			} catch (NumberFormatException e) {
				throw new IllegalExchangeMessageData(
						"illegal statement for price");
			}
		}

		ticker = clientMessage.getSymbol();
		s = clientMessage.getSide();
		quantity = clientMessage.getQuantity();
		synchronized (exchange) {
			orderId = exchange.size() + 1;
		}
		String userLogin = "";

		for (UserDataBasa user : tableOfUsers) {
			if (user.getKey().equals(message.getKey())) {
				userLogin = user.getLogin();
				break;
			}
		}

		if (ot == OrderType.LIMIT) {
			order = new Order(orderId, price, quantity, s, userLogin,
					clientMessage.getClientOrderId());
		} else {
			order = new Order(orderId, quantity, s, userLogin,
					clientMessage.getClientOrderId());
		}
		

		return new SymboledOrder(order, ticker);
	}

	private boolean userIsLoginedCorrectly(DefinedMessage message) {

		SelectionKey key = message.getKey();
		ExchangeMessage clientMessage = message.getMessage();

		try {
			if (clientMessage.getMessageType() != MessageType.LOGON) {
				// ERROR

				String rejectReason = "the first message is not a login";
				clientMessage = MessageConstructor
						.constructRejectMessage(rejectReason);

				synchronized ((SocketChannel) key.channel()) {

					mt.transmit(clientMessage.toByteArray(),
							(SocketChannel) key.channel());
				}
				// disconnect this user
				key.cancel();
				return false;
			} else { // MessageType = LOGON
				if (usersLogins.contains(clientMessage.getUsername())) {
					// such login is already been used

					String rejectReason = "this login is already been taken";
					clientMessage = MessageConstructor
							.constructRejectMessage(rejectReason);

					synchronized ((SocketChannel) key.channel()) {

						mt.transmit(clientMessage.toByteArray(),
								(SocketChannel) key.channel());
					}
					// disconnect this user
					key.cancel();
					return false;
				} else { // login is correct
					loginedClients.add(key);
					synchronized (tableOfUsers) {
						tableOfUsers.add(new UserDataBasa(key, clientMessage
								.getUsername()));
					}
					synchronized (usersLogins) {
						usersLogins.add(clientMessage.getUsername());
					}
					synchronized ((SocketChannel) key.channel()) {
						mt.transmit(message.getMessage().toByteArray(),
								(SocketChannel) key.channel());
					}
					return true;

				}
			}
		} catch (IOException e) {
			System.out.println("server failed to transmit a message");
			return false;
		}
	}

	private void processMatching(Order order, String ticker, SelectionKey key) {

		Queue<ExchangeMessage> reportsForClients = new LinkedList<ExchangeMessage>();
		ExchangeMessage em;
		String userLogin;
		OpenBook stock = exchange.get(ticker);
		stock.matchOrder(order, reportsForClients);
		while (!reportsForClients.isEmpty()) {
			em = reportsForClients.remove();			
			try {
				userLogin = em.getUsername();
				if (userLogin.equals(order.getClientLogin())) {					
					mt.transmit(em.toByteArray(), (SocketChannel) key.channel());
				} else {					
					for (UserDataBasa user : tableOfUsers) {
						if (user.getLogin().equals(userLogin)) {							
							mt.transmit(em.toByteArray(), (SocketChannel) user
									.getKey().channel());
							break;
						}
					}
				}
			} catch (IOException e) {
				System.out.println("transmition has failed");
			}
		}
		return;
	}

}
